Google Cloud Storageへストリーミングアップロードを行う

#Google Cloud Storage

#Python

kobayashi.m

2023.01.11

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

はじめに

データアナリティクス事業本部のkobayashiです。

PythonでデータをCSVなどに保存することなくメモリ上から直接Google Cloud Storage（GCS）へファイルをアップロードする場合に逐次データが増えるといったシチュエーションなどではストリーミングアップロードが使えるかと思います。今回はこのGCSのストリーミングアップロードを試したのでまとめます。

ストリーミングアップロード | Cloud Storage | Google Cloud

Google Cloud Storage（GCS）へのストリーミングアップロード

PythonでGCSへストリーミングアップロードを行うにはBlobWriterクラスを使います。BlobWriter クラスの詳しい使い方は公式ドキュメント に記載されているのですが、ストリーミングアップロードにはwriterメソッドを使います。他にもチェックサムを行うメソッドやメソッドなどストリーミング処理に使うメソッドが一式などがあります。

では早速コードを実装してストリーミングアプロードを行ってみます。

BlobWriterクラスを使ったアップロードを行う

今回GCSへのストリーミングアップロードを確認するために以下の手順でテストします。

sample.csvを用意する
sample.csvをopenして1行づつ読み込む
1行読み込んだデータをwriteメソッドで書き込む

以下が実際のコードになります。

from google.cloud import storage
from google.cloud.storage.fileio import BlobWriter

client = storage.Client()
bucket = client.get_bucket("cm-ncl-dev-jedi-data-raw-input")
blob = bucket.blob("test-sftp/sample.csv")
writer = BlobWriter(blob)

with open("./sample.csv", "rb") as fp:
for line in fp:
    writer.write(line)
writer.close()

特に難しいポイントはありません。Pythonクライアントライブラリでgoogle.cloud.storage.Clientクラスをインスタンス化してアップロード先のオブジェクトを取得し、BlobWriterクラスに渡してインスタンス化します。その後writerメソッドで逐次データを書き込み、最後にcloseメソッドで書き込み終了します。

後はこれを実行することでGCSへストリーミングアップロードが完了します。